package com.saic.bolt;

import java.util.HashMap;
import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;

import com.google.gson.Gson;
import com.saic.utils.HbaseUtil;

public class HBaseBolt extends BaseRichBolt  {

	/**
	 * 解析JSONString 为Map，map在封装到Tuple中，下一步传入HBaseBolt中
	 */
	private static final long serialVersionUID = 1L;
	private OutputCollector collector;  
	
	/**
	 * 负责解析JSONString 为 Map
	 */
	public void execute(Tuple input) {
		String word = input.getString(0);
		HbaseUtil.putData(JSONStringToMap(word));
		collector.ack(input);
	}
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		
		declarer.declare(new Fields("key"));

	}

	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		// TODO Auto-generated method stub
		this.collector = collector;
		
		
	}

	   public static Map<String,String> JSONStringToMap(String JSONString){
	        Map<String,String> map  = new HashMap<String, String>();
	        Gson gson = new Gson();
	        java.lang.reflect.Type type = new com.google.gson.reflect.TypeToken<Map<String, Object>>(){}.getType();
	        map = gson.fromJson(JSONString, type);
	        return map;

	    }
	

}
